package org.apache.rocketmq.example.MessageTestAll.Consume;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.example.MessageTestAll.Produce.quikstart_P;
import org.apache.rocketmq.example.quickstart.Producer;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class quitstart_C {

    public static AtomicInteger atomicInteger = new AtomicInteger(1);


public void Sync() throws MQClientException {
    DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(String.valueOf(atomicInteger.incrementAndGet()));
    defaultMQPushConsumer.setNamesrvAddr(Producer.DEFAULT_NAMESRVADDR);
    defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    defaultMQPushConsumer.subscribe("sync_topic","sync_tag");


    defaultMQPushConsumer.registerMessageListener(
            new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

try{
    msgs.forEach(msg-> {
        try {
            System.out.printf("接收同步消息: 主题=%s, 标签=%s, 消息ID=%s, 内容=%s%n",
                    msg.getTopic(),
                    msg.getTags(),
                    msg.getMsgId(),
                    new String(msg.getBody(), "UTF-8"));
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException();
        }
    });
}catch (RuntimeException e){
    e.printStackTrace();
    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            }
    );

defaultMQPushConsumer.start();



}

public void Async() throws MQClientException {
    DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(String.valueOf(atomicInteger.incrementAndGet()));
    defaultMQPushConsumer.setNamesrvAddr(Producer.DEFAULT_NAMESRVADDR);
    defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    defaultMQPushConsumer.subscribe("Async_topic","Async_tag");


    defaultMQPushConsumer.registerMessageListener(
            new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                    try{
                        msgs.forEach(msg-> {
                            try {
                                System.out.printf("接收同步消息: 主题=%s, 标签=%s, 消息ID=%s, 内容=%s%n",
                                        msg.getTopic(),
                                        msg.getTags(),
                                        msg.getMsgId(),
                                        new String(msg.getBody(), "UTF-8"));
                            } catch (UnsupportedEncodingException e) {
                                throw new RuntimeException();
                            }
                        });
                    }catch (RuntimeException e){
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            }
    );

    defaultMQPushConsumer.start();

}

public void Order() throws MQClientException {

    DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(String.valueOf(atomicInteger.incrementAndGet()));
    defaultMQPushConsumer.setNamesrvAddr(Producer.DEFAULT_NAMESRVADDR);
    defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    defaultMQPushConsumer.subscribe("orderly_topic","*");


    defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {
                                                      @Override
                                                      public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                                                          context.setAutoCommit(true);

                                                          List<StringBuilder> collect = msgs.stream().map(i -> new StringBuilder().append(i.getBody().toString()).append("+").append(i.getQueueId())).collect(Collectors.toList());

                                                          collect.forEach(i-> System.out.println(i.toString()));

                                                          return ConsumeOrderlyStatus.SUCCESS;
                                                      }
                                                  }
    );

    defaultMQPushConsumer.start();




}

public void Delay() throws MQClientException {
    DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(String.valueOf(atomicInteger.incrementAndGet()));
    defaultMQPushConsumer.setNamesrvAddr(Producer.DEFAULT_NAMESRVADDR);
    defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    defaultMQPushConsumer.subscribe("delay_topic","delay_tag");


    defaultMQPushConsumer.registerMessageListener(
            new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                    try{
                        msgs.forEach(msg-> {
                            try {
                                System.out.printf("接收延迟消息: 主题=%s, 标签=%s, 消息ID=%s, 内容=%s%n",
                                        msg.getTopic(),
                                        msg.getTags(),
                                        msg.getMsgId(),
                                        new String(msg.getBody(), "UTF-8"));
                            } catch (UnsupportedEncodingException e) {
                                throw new RuntimeException();
                            }
                        });
                    }catch (RuntimeException e){
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            }
    );

    defaultMQPushConsumer.start();




}

    public static void main(String[] args) {
        quitstart_C quitstartC = new quitstart_C();
        Method[] methods = quitstart_C.class.getMethods();
        for (int i = 0; i < methods.length; i++) {

            if(methods[i].getDeclaringClass() == Object.class)
                continue;

            try {
                if(methods[i].getName().equals("main"))
                {
                    continue;
                }
                methods[i].invoke(quitstartC);
            } catch (IllegalAccessException e) {
                throw new RuntimeException(e);
            } catch (InvocationTargetException e) {
                throw new RuntimeException(e);
            }
        }

        while (true);
    }


}
